package com.dy.dws.TradeHospPatGenAGPresWin;

import com.alibaba.fastjson.JSONObject;
import com.dy.dws.function.MyAsyncDataStream;
import com.dy.realtime.common.base.BaseApp;
import com.dy.realtime.common.bean.DwsTradeHospPatGenAGPresBean;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.DateFormatUtil;
import com.dy.realtime.common.util.DorisMapFunction;
import com.dy.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.dws.TradeHospPatGenAGPresWin
 * @className: DwsTradeHospPatGenAGPresWin
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/14 14:29
 * @version: 1.0
 * 交易域医院性别年龄段粒度处方开单各窗口汇总表
 */

public class DwsTradeHospPatGenAGPresWin extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeHospPatGenAGPresWin().start(Constant.TOPIC_DWD_TRADE_PRESCRIPTION, Constant.TOPIC_DWS_TRADE_HOSP_PAT_GEN_A_G_PRES_WIN, 4, 10021);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        SingleOutputStreamOperator<JSONObject> eltStream = getEltStream(dataStreamSource);
//        eltStream.print();
        //转换为实体类对象
        SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> beanStream = getBeanStream(eltStream);
//        beanStream.print();
        //获取hosptitalId
        SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> getHospitalId = getHospitalId(beanStream);
//        getHospitalId.print();
        //获取年龄段与性别
//        getHospitalName.print();
        SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> gender = getGender(getHospitalId);
//        gender.print();
        SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> reduce = getReduce(gender);
        //获取hosptitalName

        SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> getHospitalName = getHospitalName(reduce);
//        getHospitalName.print();
        SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> result = AsyncDataStream.unorderedWait(getHospitalName, new MyAsyncDataStream<DwsTradeHospPatGenAGPresBean>("dim_dict") {
            @Override
            public void getResult(DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean, JSONObject jsonObject) {
                dwsTradeHospPatGenAGPresBean.setPatientGender(jsonObject.getString("value"));
            }

            @Override
            public String getDimId(DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean) {
                return dwsTradeHospPatGenAGPresBean.getPatientGenderCode();
            }
        }, 100, TimeUnit.MINUTES, 20);
//        result.print();
        result.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_hosp_pat_gen_a_g_pres_win"));


    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> getReduce(SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> gender) {
        return gender.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsTradeHospPatGenAGPresBean>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<DwsTradeHospPatGenAGPresBean>() {
                    @Override
                    public long extractTimestamp(DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean, long l) {
                        return dwsTradeHospPatGenAGPresBean.getTs();
                    }
                })).keyBy(new KeySelector<DwsTradeHospPatGenAGPresBean, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> getKey(DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean) throws Exception {
                        return Tuple3.of(dwsTradeHospPatGenAGPresBean.getHospitalId(), dwsTradeHospPatGenAGPresBean.getAgeGroup(), dwsTradeHospPatGenAGPresBean.getPatientGenderCode());
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(5))).
                reduce(new ReduceFunction<DwsTradeHospPatGenAGPresBean>() {
                    @Override
                    public DwsTradeHospPatGenAGPresBean reduce(DwsTradeHospPatGenAGPresBean t1, DwsTradeHospPatGenAGPresBean t2) throws Exception {
                        t1.setPrescriptionAmount(t1.getPrescriptionAmount().add(t2.getPrescriptionAmount()));
                        t1.setPrescriptionCount(t1.getPrescriptionCount() + t2.getPrescriptionCount());
                        return t1;
                    }
                }, new ProcessWindowFunction<DwsTradeHospPatGenAGPresBean, DwsTradeHospPatGenAGPresBean, Tuple3<String, String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple3<String, String, String> stringStringStringTuple3, ProcessWindowFunction<DwsTradeHospPatGenAGPresBean, DwsTradeHospPatGenAGPresBean, Tuple3<String, String, String>, TimeWindow>.Context context, Iterable<DwsTradeHospPatGenAGPresBean> iterable, Collector<DwsTradeHospPatGenAGPresBean> collector) throws Exception {
                        for (DwsTradeHospPatGenAGPresBean bean : iterable) {
                            String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                            String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                            String curDate = DateFormatUtil.tsToDate(System.currentTimeMillis());

                            bean.setStt(stt);
                            bean.setEdt(edt);
                            bean.setCurDate(curDate);
                            collector.collect(bean);
                        }
                    }
                });
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> getGender(SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> getHospitalName) {
        return AsyncDataStream.unorderedWait(getHospitalName, new MyAsyncDataStream<DwsTradeHospPatGenAGPresBean>("dim_patient") {
            @Override
            public void getResult(DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean, JSONObject jsonObject) {
                String birthday = jsonObject.getString("birthday");
                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
                String s = DateFormatUtil.tsToDate(System.currentTimeMillis());
                LocalDate parse = LocalDate.parse(s, formatter);
                int year1 = parse.getYear();
                int age;
                String ageGroup = null;
                if (birthday != null) {
                    LocalDate birthDate = LocalDate.parse(birthday, formatter);
                    int year = birthDate.getYear();
                    age = year1 - year;
                    if (age >= 0 && age <= 2) {
                        ageGroup = "婴儿期";
                    } else if (age <= 5) {
                        ageGroup = "幼儿期";
                    } else if (age <= 11) {
                        ageGroup = "小学阶段";
                    } else if (age <= 17) {
                        ageGroup = "青少年期";
                    } else if (age <= 29) {
                        ageGroup = "青年期";
                    } else if (age <= 59) {
                        ageGroup = "中年期";
                    } else if (age > 60) {
                        ageGroup = "老年期";
                    } else {
                        ageGroup = "年龄异常";
                    }
                }
                dwsTradeHospPatGenAGPresBean.setAgeGroup(ageGroup);
                dwsTradeHospPatGenAGPresBean.setPatientGenderCode(jsonObject.getString("gender"));
            }

            @Override
            public String getDimId(DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean) {
                return dwsTradeHospPatGenAGPresBean.getPatientId();
            }
        }, 100, TimeUnit.MINUTES, 20);
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> getHospitalName(SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> getHospitalId) {
        return AsyncDataStream.unorderedWait(getHospitalId, new MyAsyncDataStream<DwsTradeHospPatGenAGPresBean>("dim_hospital") {
            @Override
            public void getResult(DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean, JSONObject jsonObject) {
                dwsTradeHospPatGenAGPresBean.setHospitalName(jsonObject.getString("name"));
            }

            @Override
            public String getDimId(DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean) {
                return dwsTradeHospPatGenAGPresBean.getHospitalId();
            }
        }, 100, TimeUnit.MINUTES, 20);
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> getHospitalId(SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> beanStream) {
        return AsyncDataStream.unorderedWait(beanStream, new MyAsyncDataStream<DwsTradeHospPatGenAGPresBean>("dim_doctor") {
            @Override
            public void getResult(DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean, JSONObject jsonObject) {
                dwsTradeHospPatGenAGPresBean.setHospitalId(jsonObject.getString("hospital_id"));
            }

            @Override
            public String getDimId(DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean) {
                return dwsTradeHospPatGenAGPresBean.getDoctorId();
            }
        }, 100, TimeUnit.MINUTES, 20);
    }

    private static SingleOutputStreamOperator<DwsTradeHospPatGenAGPresBean> getBeanStream(SingleOutputStreamOperator<JSONObject> eltStream) {
        return eltStream.map(new MapFunction<JSONObject, DwsTradeHospPatGenAGPresBean>() {
            @Override
            public DwsTradeHospPatGenAGPresBean map(JSONObject jsonObject) throws Exception {
                DwsTradeHospPatGenAGPresBean dwsTradeHospPatGenAGPresBean = JSONObject.parseObject(jsonObject.toString(), DwsTradeHospPatGenAGPresBean.class);
                dwsTradeHospPatGenAGPresBean.setPrescriptionAmount(dwsTradeHospPatGenAGPresBean.getTotalAmount());
                dwsTradeHospPatGenAGPresBean.setPrescriptionCount(1L);
                return dwsTradeHospPatGenAGPresBean;
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> getEltStream(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(s);
                    String doctor_id = jsonObject.getString("doctor_id");
                    String patient_id = jsonObject.getString("patient_id");
                    if (doctor_id != null && patient_id != null) {
                        collector.collect(jsonObject);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
}